package com.cscec8b.ct.consumer.dao;

import com.cscec8b.ct.common.constant.Names;
import com.cscec8b.ct.common.dao.BaseDao;
import com.cscec8b.ct.consumer.bean.CallLog;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @Copyright: Shanghai Definesys Company.All rights reserved.
 * @Description:
 * @author: chuhaitao
 * @since: 2019/1/27 20:38
 * @history: 1.2019/1/27 created by chuhaitao
 */
public class HBaseDao extends BaseDao {


    /**
     * 初始化  创建命名空间，创建表
     *
     * @throws Exception
     */
    public void init() throws Exception {
        start();
        //创建表名
        createNameSpaceNX(Names.NAMESPACE.getValue());
        //预分区
        createTableXX(Names.TABLE.getValue(), 6, "com.cscec8b.ct.consumer.coproessor.CalleeProessor", Names.CF_CALLER.getValue());

        end();
    }

    /**
     * 1.1版本 插入数据到HBase
     *
     * @param value
     */
    public void insertData(String value) throws Exception {

        String[] values = value.split("\t");


        String call1 = values[0];
        String call2 = values[1];
        String callTime = values[2];
        String duration = values[3];
        //rowKey
        int regionNum = getRegionNum(call1, callTime);

        String rowKey = regionNum + "_" + call1 + "_" + callTime + "_" + call2 + "_" + duration;
        //创建put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        //设置列族
        byte[] family = Bytes.toBytes(Names.CF_CALLER.getValue());
        //设置列,列族/列名/值
        put.addColumn(family, Bytes.toBytes("call1"), Bytes.toBytes(call1));
        put.addColumn(family, Bytes.toBytes("call2"), Bytes.toBytes(call2));
        put.addColumn(family, Bytes.toBytes("callTime"), Bytes.toBytes(callTime));
        put.addColumn(family, Bytes.toBytes("duration"), Bytes.toBytes(duration));


        //插入数据
        putData(Names.TABLE.getValue(), put);
    }

/*
*  5_19683537146_20180423195725_17885275338_0
*  column=caller:duration, timestamp=1549691226826, value=0171
     5_19683537146_201804    5_19683537146_201804|
* */

    /**
     * 1.2 使用对象的方式   优化代码
     *
     * @param log
     * @throws Exception
     */
    public void insertData(CallLog log) throws Exception {
        //rowKey
        int regionNum = getRegionNum(log.getCall1(), log.getCallTime());
        String rowKey = regionNum + "_" + log.getCall1() + "_" + log.getCallTime() + "_" + log.getCall2() + "_" + log.getDuration() + "_" + "1";
        //0_13956643892_20190211_13935457872_30000
        log.setRowKey(rowKey);
        putData(log);
    }

    //0_1333_20180901_   ~  0_1333_20180901_
    //1_1333_20180901_   ~  1_1333_20180901_
    //获取分区


}
